package com.atguigu.chapter02;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/8 10:03
 */
public class Flink01_Batch_WordCount {
    public static void main(String[] args) throws Exception {
        // 批处理
        
        // 1. 创建一个批处理的执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 2. 通过执行环境, 得个DataSet
        DataSource<String> lineDS = env.readTextFile("input/words.txt");
        // 3. 做各种转换
        FlatMapOperator<String, String> wordDS = lineDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line,
                                Collector<String> collector) throws Exception {
                for (String word : line.split(" ")) {
                    collector.collect(word);
                }
            }
        });
        // "hello"-> (hello, 1) "word"->(world, 1)
        MapOperator<String, Tuple2<String, Long>> wordAndOneDS = wordDS.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String word) throws Exception {
                return Tuple2.of(word, 1L);
            }
        });
        
        // 按照元组的下标进行分组
        UnsortedGrouping<Tuple2<String, Long>> groupedDS = wordAndOneDS.groupBy(0);
        
        // 按照元组下标进行聚合
        AggregateOperator<Tuple2<String, Long>> result = groupedDS.sum(1);
        
        // 4. 输出
        result.print();
        
    }
}
/*
spark
  1. 创建sparkContext
  
  2. 获取一个rdd
  
  3. 做各种转换
  
  4. 行动算子
  
  5. 关闭sparkContext

 */
